package com.duowan.realtime.scheduled.batch.reader;

import com.duowan.common.util.Profiler;
import com.duowan.realtime.dao.redis.RedisDao;
import com.duowan.realtime.scheduled.util.MVELUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.listener.ItemListenerSupport;
import org.springframework.batch.item.*;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;

import java.util.*;

/**
 * @version 1.0
 * @author: tangsicheng
 * @date: 14-9-5
 * @since 1.0
 */
public class RedisKVReader extends ItemListenerSupport implements ItemStreamReader<Map>, InitializingBean {

    private static final Logger logger = LoggerFactory.getLogger(RedisKVReader.class);

    protected RedisDao redisDao;

    protected String redisKeySetName;


    private Map resultRedisKVMap = null;

    public void setRedisKeySetName(String keySetName) {
        this.redisKeySetName = keySetName;
    }

    public void setRedisDao(RedisDao redisDao) {
        this.redisDao = redisDao;
    }

    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        resultRedisKVMap = new HashMap();
        if (StringUtils.isNotBlank(redisKeySetName)) {
            String evaluedKey = evalKeyPattern(redisKeySetName);
            resultRedisKVMap = readerFromKeySet(evaluedKey);
        }
    }

    private Map readerFromKeySet(String keySetName) {
        Profiler.start();
        Profiler.enter("读取" + keySetName + "中的redisKey");

        Map result = Maps.newHashMap();


        Set<String> keySet = redisDao.getRedisTemplate().smembers(keySetName);
        List<String> keyList = Lists.newArrayList(keySet);

        Profiler.enter("read from redis");
        List<String> valueList = redisDao.multiGet(keyList);
        Profiler.release(keyList.size());

        if (CollectionUtils.isEmpty(keyList) || CollectionUtils.isEmpty(valueList)) return result;
        Assert.isTrue(keyList.size() == valueList.size(), "查询key的返回结果大小不一致");
        int index = 0;
        for (String value : valueList) {
            String key = keyList.get(index++);
            result.put(key, value);
        }

        Profiler.release(result.size());
        Profiler.release();
        logger.info(Profiler.dump());
        return result;
    }

    @Override
    public void afterRead(Object item) {
        resultRedisKVMap = null;
    }

    @Override
    public Map read() throws UnexpectedInputException, ParseException, NonTransientResourceException {
        return resultRedisKVMap;
    }

    protected List<String> evalueKeys(List<String> redisKeys) {
        List<String> result = new LinkedList<String>();
        for (String redisKey : redisKeys) {
            String evaluedRedisKey = evalKeyPattern(redisKey);
            result.add(evaluedRedisKey);
        }
        return result;
    }

    //TODO 占位符应该更加灵活一点，支持properties
    private String evalKeyPattern(String originKeyPattern) {
        if (originKeyPattern.contains("@")) {
            String mvelKeyPattern = removeBracke(originKeyPattern);
            Map var = new HashMap();
            var.put("today", new Date());
            return (String) MVELUtil.eval(mvelKeyPattern, var);
        } else {
            return originKeyPattern;
        }

    }

    private String removeBracke(String str) {
        return str.replaceAll("@\\{", "").replace("}", "");
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        //  Assert.notEmpty(redisKeys, "请指定要查找的redis key");
    }


    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {
    }

    @Override
    public void close() throws ItemStreamException {

    }

}
